package com.udf.flink.scala.apitest.cep
import java.util

import org.apache.flink.cep.PatternSelectFunction
import org.apache.flink.cep.scala.{CEP, PatternStream}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.cep.scala.pattern.Pattern
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.windowing.time.Time
case class Record(classId:String, name:String, age:Int)
object CEPTest extends App {
  // 创建执行环境
  val env = StreamExecutionEnvironment.getExecutionEnvironment
  //这里是设置自动生成水印的时间间隔 因为由于大数据数据量很大所以每一个数据都生成一个水印会浪费性能 况且很多时候一批数据中有很多数据时间是一样的
  // 所以这里都是用周期性的生成水印 默认是200毫秒这里也可以根据自身设定
//  env.getConfig.setAutoWatermarkInterval(20000L)
//  env.setParallelism(1)
//  env.setStreamTimeCharacteristic()

  // 1. 获取数据输入流
  val stream1: DataStream[String] = env.socketTextStream("localhost",9999)

    val value: DataStream[Record] = stream1
    .map(data => {
      val arr = data.split(",")
      Record(arr(0), arr(1), arr(2).toInt)
    })
  // 2. 定义一个 Pattern， 匹配两秒内处理到的年龄都是 20 岁的两个人
  private val pattern: Pattern[Record, Record] = Pattern
    .begin[Record]("start").where(_.age == 20) // 匹配第一个获取到的年龄是 20 岁
    .next("next").where(_.age == 20)// 匹配第二个获取到的年龄是 20 岁
    .within(Time.minutes(2)) // 设定约束时间为 2 秒
  // 3. 将创建好的 Pattern 应用到输入事件流上
  private val patternStream: PatternStream[Record] = CEP.pattern[Record](value, pattern)

  // 4. 获取事件序列，得到匹配到的数据 注意这里inProcessTime 用于指定process
  private val result: DataStream[String] = patternStream.inProcessingTime().select(new PatternMacthData2)
  result.print("CEPTest")
  env.execute()
}

/**
 * 定义一个Pattern Select Function 对匹配到的数据进行处理
 */
class PatternMacthData2 extends PatternSelectFunction[Record,String]{
  override def select(pattern: util.Map[String, util.List[Record]]): String = {
    // 从 map 中根据名称获取对应的事件
    val start: Record = pattern.get("start").iterator().next()
    val next: Record = pattern.get("next").iterator().next()
    s"${start.name} 与 ${next.name} 都是 ${start.age} 岁"
  }
}
/*

   val env = StreamExecutionEnvironment.getExecutionEnvironment
    //   val value = env.socketTextStream("localhost",7777)
    val value = env.addSource(new PerSource)

    env.setRestartStrategy(RestartStrategies.noRestart())
    //每15秒内 每个传感器的最小值
    env.setRestartStrategy(RestartStrategies.noRestart())
    //处理的类型 时流还是批处理  还是自动 建议不要去设置
    // env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC)
    //这里是设置自动生成水印的时间间隔 因为由于大数据数据量很大所以每一个数据都生成一个水印会浪费性能 况且很多时候一批数据中有很多数据时间是一样的 //所以这里都是用周期性的生成水印 默认是200毫秒这里也可以根据自身设定
    env.getConfig.setAutoWatermarkInterval(200L)
    //这里最新的flink 是事件时间为默认项
    //这里就是设置所要处理事件的类型 事件时间 处理时间  数据进入时间 如果是处理时间那么就不需要设置太多因为本身就是自身处理的时间但是出现乱序后 会影响 数据最终的准确性
   // env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime) //处理时间
    val value1 = value
      //必须加上 flink
		//这里就是分配时间戳和水印了  分配时间戳的意思就是 此次处理数据是按照哪个字段为时间 来进行处理 而 水印就是延迟时间去关闭桶 这里有很多设置一般都是使用过 flink自己封装好的方法  比如BoundedOutOfOrdernessTimestampExtractor
      /*.assignTimestampsAndWatermarks(new timestamps.BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.seconds(1)) {
        override def extractTimestamp(element: SensorReading): Long = {
          element.timestamp
        }
      })*/
      // .assignTimestampsAndWatermarks(new waterMarkDeom(Time.seconds(3)))
     //如果使用最新的会发现 上面的使用方式 已经被标记了 推荐是用过 WatermarkStrategy 水印的策略 来进行设置 ， 这里 flink 也封装了三个方法 为我们所用1）1）1)forBoundedOutOfOrderness 就是平时用的有界输出数据 2）.forMonotonousTimestamps()自增长输出 这里自增长大概意思就是事件时间没有乱序都是正确的 不需要去设置水印（有不对的地方欢迎指正 目前也是正在学习） 3） noWatermarks 这个就不用翻译了大家都懂 然后需要在后面添加事件时间的取自哪个字段也是 flink自己封装的方法.withTimestampAssigner 时间戳分配器 其中有两个重写的方法 一个是添加 SerializableTimestampAssigner 另一个 TimestampAssignerSupplier 这里我只用了第一个第二个没有研究 等有懂的大佬 来评论告诉下 然后 最后还一个.withIdleness(Duration.ofMillis(1))) 这个方法可添加也可以不添加 就不会影响启动了 翻译的大概意思是就是flink 为分布式 处理流式框架 在由上游数据发送下游的时候需要等到所有分区的数据到达后 以最后的数据到达时间为时间但是有可能出现超时会影响下一步操作所以这个配置了之后大概的意思是保留空闲分区的时间之后将不会影响向下游广播 数据操作流程
     .assignTimestampsAndWatermarks(WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofMillis(1 )).withTimestampAssigner(new SerializableTimestampAssigner[SensorReading]() {
         override def extractTimestamp(element: SensorReading, recordTimestamp: Long): Long = {
           element.timestamp*1000L
         }
       }).withIdleness(Duration.ofMillis(1)))
        /*.assignTimestampsAndWatermarks(WatermarkStrategy.forMonotonousTimestamps().withTimestampAssigner(new SerializableTimestampAssigner[SensorReading]() {
          override def extractTimestamp(element: SensorReading, recordTimestamp: Long): Long = {
            element.timestamp*1000L
          }
        }))*/
      /*.assignTimestampsAndWatermarks(WatermarkStrategy.forGenerator(new waterMarkStrategyDemo).withTimestampAssigner(new SerializableTimestampAssigner[SensorReading] {
        override def extractTimestamp(element: SensorReading, recordTimestamp: Long): Long = {
          element.timestamp
        }
      }).withIdleness(Duration.ofSeconds(1)))*/
      //WatermarkStrategy.forGenerator(new waterMarkStrategyDemo)
    /*  .assignTimestampsAndWatermarks(WatermarkStrategy.forGenerator(new waterMarkStrategyDemo)withTimestampAssigner(new SerializableTimestampAssigner[SensorReading] {
        override def extractTimestamp(element: SensorReading, recordTimestamp: Long): Long = {
          element.timestamp
        }
      }))*/
     /* .assignTimestampsAndWatermarks(WatermarkStrategy.forMonotonousTimestamps().withTimestampAssigner((new SerializableTimestampAssigner[SensorReading]() {
        override def extractTimestamp(element: SensorReading, recordTimestamp: Long): Long = {
          element.timestamp
        }
      })))*/
     //排好序的升序的 数据 来的时间 不用定义watermark 数据没有乱序
       //.assignAscendingTimestamps(x=>x.timestamp*1000)
      .keyBy(_.id)
      //.timeWindow(Time.seconds(12)).reduce((x, y) => x)
      //滚动窗口 头尾相接、
      .timeWindow(Time.seconds(1))
      //这个设置时允许时间窗口处理过后 保留窗口的时间 大概的意思比如 0-1秒的窗口操作了但是保留1毫秒 然后在这1毫秒内 如果还有0-1秒的数据到来也会参与到数据的操作
      .allowedLateness(Time.milliseconds(1))
      //.aggregate(AggregationType.MAX,2)
      //输出哪个数据延迟到来 这就是将保留窗口 都已经关闭了 0-1秒的数据 来了 就会进入到这里测流输出 在之后可用value1.getSideOutput(new OutputTag[SensorReading]("later")) 接住 然后 可以在处理比如 发送到一个 延迟数据 topic 然后 跟以前的业务数据 操作 来避免数据丢失 比如最小值 那么跟 最新的数据取比对 如果 比最新的数据要小 那么就更新如果不是 就可以抛弃了
      .sideOutputLateData(new OutputTag[SensorReading]("later"))
      //.window(TumblingProcessingTimeWindows.of(Time.seconds(1)))//滑动时间窗口
      // .window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))

     .reduce((x, y) => x)
    //.window(EventTimeSessionWindows.withGap(Time.seconds(1)))//会话时间窗口  1s进来的都算一个桶
    // .timeWindow()准备弃除
    //  .countWindow(10)
    // .trigger(new sw)
    // .evictor(new sa)
    // .allowedLateness(Time.seconds(1))
    // .sideOutputLateData(new OutputTag("ds"))
    // unit.getSideOutput(new OutputTag("ds"))
    value1.print("ok")
    val value2: DataStream[SensorReading] = value1.getSideOutput(new OutputTag[SensorReading]("later"))

    env.execute("ss ss")

 */